package com.lunatic.rocketmq.demo;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.io.IOException;

public class TransactionProducer {


    /**
     * 事务消息原理
     * 事务消息采用两阶段提交模式：
     *
     * 发送预处理消息：生产者先发送一个半消息（Half Message）到 Broker，此时消息对消费者不可见
     * 执行本地事务：Broker 接收半消息后，生产者执行本地事务
     * 提交事务状态：根据本地事务执行结果（提交或回滚），生产者向 Broker 发送二次确认
     * 消息状态回查：若 Broker 未收到确认，会定期查询生产者事务状态
     *
     * @param args
     */
    public static void main(String[] args) throws MQClientException, IOException {
        TransactionMQProducer producer = new TransactionMQProducer("txt_producer_1");
        producer.setNamesrvAddr(Constants.NAME_SRV_ADDR);
        producer.start();

        // 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 执行本地事务
             * @param message 半消息
             * @param o 额外参数
             * @return 本地事务执行状态
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                String tags = message.getTags();
                return switch (tags) {
                    case "tag1" -> LocalTransactionState.COMMIT_MESSAGE;  // 提交事务，消息将对消费者可见
                    case "tag2" -> LocalTransactionState.ROLLBACK_MESSAGE;  // 回滚事务，消息将被丢弃
                    default -> LocalTransactionState.UNKNOW;     // 未知状态，需要 broker 回查
                };
            }

            /**
             * 事务状态回查
             * @param messageExt 待查询的消息
             * @return 本地事务状态
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("=============check " + messageExt);
                // 假设经过查询，所有未知状态的事务最终都提交
                return LocalTransactionState.UNKNOW;
            }
        });

        for (int i = 0; i < 10; i++) {
            Message message = new Message();
            if (i % 2 == 0) {
                message.setTags("tag1");
            } else {
                message.setTags("tag2");
            }
            message.setTopic("txtTopic");
            message.setBody((i + " hello. 世界").getBytes());
            TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, null);
            System.out.println(transactionSendResult);
        }
        Message message = new Message();
        message.setTags("tag3");
        message.setTopic("txtTopic");
        message.setBody(("hello.").getBytes());
        TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, null);
        System.out.println(transactionSendResult);

    }
}
